package com.dr.cmdb.core.kafka.consumer;


import com.dr.cmdb.common.log.LoggerUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @ProjectName: cmdb
 * @Package: com.dr.cmdb.application.kafka.consumer
 * @Description: java类作用描述
 * @Author: duanrui
 * @CreateDate: 2021/4/14 18:52
 * @Version: 1.0
 * <p>
 * Copyright: Copyright (c) 2021
 */
@Component
@Slf4j
public class KafkaReceiver {

    @KafkaListener(topics = {"test"},groupId = "c1")
    public void listen(ConsumerRecord<?, ?> record, Consumer consumer) {

        Optional<?> kafkaMessage = Optional.ofNullable(record.value());

        if (kafkaMessage.isPresent()) {

            Object message = kafkaMessage.get();
            LoggerUtil.error(log,"日志信息",record.toString());
            LoggerUtil.error(log,message.toString());
        }
        consumer.commitAsync();

    }
}
